"""
    基本面量化
    * 业绩超预期因子
        - 业绩预期计算方法：线性漂移
    * 盈利偏度
"""
import datetime
import numpy as np
import pandas as pd
from data_resource.data_bases import engine
from utilities.utilities_func import sliding_window, bulk_insert_with_orm, get_next_tradeDate
from utilities.utilities_statics import align_trade_date, factor_standardize, factor_neutralize
from factorAnaly.financialEngine import FinancialFactor

"""
    通用函数
"""


def financial_quantile(group):
    """每交易日截面计算财务因子分位数值"""
    ranks = group.rank()
    r = ranks / len(ranks)
    return r


def financial_pivot(group, n=756):
    """计算财务因子估值中枢"""
    if len(group) <= n:
        return pd.Series(np.nan, index=group.index)
    else:
        # 使用扩展窗口rank计算每个点的分位数
        _pivot = group.rolling(window=n).apply(lambda x: x.mean() / x.std())
        return _pivot


# 股息率因子
def dividend_yield():
    """
        股息率因子
    """
    _sql = """
    select a.trade_date as trading, a.code, a.dv_ratio as signal, b.close, b.open
    from quant_research.indicator_daily as a
    left join quant_research.market_daily_ts as b
    on a.code=b.ticker and a.trade_date=b.trade_date
    where a.trade_date > '2015-01-01'
    order by a.trade_date
    """

    _raw = pd.read_sql(_sql, engine)

    # 剔除股息率为0的个股

    return _raw[_raw['signal'] != 0]


# 盈利偏度因子, 研报中因子测试效果不佳。
def profitability_skewness():
    pass


# EP综合因子 (分层效果弱, 需要优化股票池后再测试有效性)
def ep_compound(meager_profit=False):
    """
        PE估值合成因子 = EP_Q + EP_quantile + EP_pivot
        EP_Q: 季度EP值
        EP_quantile: 个股时间序列分位数因子选股能力弱，采用截面分位数排序值
        EP_pivot: EP估值中枢，过去三年的EP均值/EP标准差

        微利股处理：归母净利润TTM<1000万

    """

    # 获取原始PE数据
    _sql = """
    select a.trade_date, a.code, 1/a.pe as ep
    from quant_research.indicator_daily as a
    left join quant_research.market_daily_ts as b 
        on a.code=b.ticker and a.trade_date=b.trade_date
    where a.pe is not null and a.trade_date>'2020-01-01'
    order by a.trade_date
    """

    _raw = pd.read_sql(_sql, engine)
    raw = _raw.copy()

    if meager_profit:
        # 获取个股净利润ttm数据做为标签
        _sql1 = """
        select iq.ticker as code, iq.f_ann_date, iq.end_date, iq.n_income_attr_p
                from (
                    select *, row_number() over (
                        partition by ticker, end_date
                        order by update_flag desc, f_ann_date
                    ) as rn
                    from quant_research."financials_incomeState_quarter"
                ) iq
        where iq.rn=1 and iq.revenue>0
        order by iq.f_ann_date
        """
        _profit = pd.read_sql(_sql1, engine)
        _profit['income_attr_ttm'] = (
            _profit.groupby('code')['n_income_attr_p']
            .transform(lambda x: x.astype(float).rolling(4).mean())
        )  # 强制转换为浮点数

        # 对齐交易日
        _profit = align_trade_date(_profit, engine, signal_columns=['n_income_attr_p', 'income_attr_ttm'])
        _profit.drop(columns=['is_open'])
        _profit.rename(columns={'trading': 'trade_date'}, inplace=True)
        # 左连接
        raw = _raw.merge(_profit, how='left', on=['trade_date', 'code'])

    # 计算截面分位数
    raw['ep_quantile'] = raw.groupby('trade_date', group_keys=False)['ep'].apply(lambda x: financial_quantile(x))

    # 计算ep估值中枢
    raw['ep_pivot_3y'] = raw.groupby('code', group_keys=False)['ep'].apply(lambda x: financial_pivot(x, 750))
    raw['ep_pivot_1y'] = raw.groupby('code', group_keys=False)['ep'].apply(lambda x: financial_pivot(x, 250))

    # bulk_insert_with_orm(raw, 'factor_ep_compound', engine)
    return raw


# BP综合因子 (分层效果弱, 需要优化股票池后再测试有效性)
def bp_compound():
    """
        PE估值合成因子 = BP_Q + BP_quantile + BP_pivot
        EP_Q: 季度EP值
        EP_quantile: 过去n个交易日作为窗口，计算EP分位数，每日变动。分位数值极度右偏，普涨个股数量非常少
        EP_pivot: EP估值中枢，过去三年的EP均值/EP标准差
    """

    # 获取原始PE数据
    _sql = """
    select a.trade_date, a.code, 1/a.pb as bp
    from quant_research.indicator_daily as a
    left join quant_research.market_daily_ts as b 
        on a.code=b.ticker and a.trade_date=b.trade_date
    where a.trade_date > '2020-01-01' and a.pb is not null
    order by a.trade_date
    """

    _raw = pd.read_sql(_sql, engine)

    # 计算截面分位数
    _raw['bp_quantile'] = _raw.groupby('trade_date', group_keys=False)['bp'].apply(lambda x: financial_quantile(x))

    # 计算ep估值中枢
    _raw['bp_pivot_3y'] = _raw.groupby('code', group_keys=False)['bp'].apply(lambda x: financial_pivot(x, 750))
    _raw['bp_pivot_1y'] = _raw.groupby('code', group_keys=False)['bp'].apply(lambda x: financial_pivot(x, 250))

    # bulk_insert_with_orm(_raw, 'factor_bp_compound', engine)

    return _raw


# 自由现金流因子，测试后无效
def cfp_compound():
    """
        CFP: FCFE / MV, FCFE=现金流量净额 + 筹资活动净额+分配股利支持的现金
        CFP估值合成因子 = CFP_Q + CFP_quantile + CFP_pivot
        CFP_Q: 季度EP值 = 现金流量净额 + 融资活动流量净额 - 支持红利现金支出
        EP_quantile: 个股时间序列分位数因子选股能力弱，采用截面分位数排序值
        EP_pivot: EP估值中枢，过去三年的EP均值/EP标准差

    """
    _sql1 = """
    with iq_cashflow as (
        select iq.ticker as code, iq.f_ann_date, iq.end_date, iq.n_cashflow_act, iq.n_cash_flows_fnc_act, iq.c_pay_dist_dpcp_int_exp
        from (
                select *, row_number() over (
                    partition by ticker, end_date
                        order by update_flag desc, f_ann_date
                    ) as rn
                from quant_research.financials_cashflow_quarter
            ) iq
        where iq.rn=1)
        
    select a.code, a.f_ann_date, a.end_date, 
        ((a.n_cashflow_act + a.n_cash_flows_fnc_act - a.c_pay_dist_dpcp_int_exp) / b.total_mv) as fcfe
    from iq_cashflow as a
    left join quant_research.indicator_daily as b on a.code=b.code and a.f_ann_date=b.trade_date
    order by a.f_ann_date
    """
    _raw = pd.read_sql(_sql1, engine)

    # 计算因子
    _raw['fcfe_ttm'] = _raw.groupby('code')['fcfe'].transform(lambda x: x.astype(float).rolling(4).mean())

    # 对齐交易日
    raw = align_trade_date(_raw, engine, signal_columns=['fcfe', 'fcfe_ttm'])
    raw.drop(columns=['is_open'])
    raw.rename(columns={'trading': 'trade_date'}, inplace=True)

    raw['fcfe_quantile'] = raw.groupby('trade_date', group_keys=False)['fcfe'].apply(lambda x: financial_quantile(x))

    # 计算ep估值中枢
    raw['fcfe_pivot_3y'] = raw.groupby('code', group_keys=False)['fcfe'].apply(lambda x: financial_pivot(x, 750))
    raw['fcfe_pivot_1y'] = raw.groupby('code', group_keys=False)['fcfe'].apply(lambda x: financial_pivot(x, 250))

    bulk_insert_with_orm(raw, 'factor_cfp_compound', engine)
    return


# EP因子估值中枢+方向优化（测试后，无效）
def ep_directedPivot(n, shift):
    """
        ep：mean/std刻画趋势稳定性，不包含趋势
        趋势信号捕捉：mean() - mean(shift)
    """

    # 获取原始PE数据
    _sql = """
    select a.trade_date as trading, a.code, 1/a.pe as ep, b.close, b.open
    from quant_research.indicator_daily as a
    left join quant_research.market_daily_ts as b on a.code=b.ticker and a.trade_date=b.trade_date
    where a.trade_date > '2010-01-01' and a.pe is not null
    order by a.trade_date
    """

    _raw = pd.read_sql(_sql, engine)

    def _trendPe(group, n, shift):
        if len(group) <= (n + shift):
            return pd.Series(np.nan, index=group.index)
        else:
            _mean = group.rolling(n).mean()
            _std = group.rolling(n).std()
            _trend = _mean - _mean.shift(shift)
            _r = np.sign(_trend) * (_mean / _std)
            return _r

    _raw['signal'] = _raw.groupby('code', group_keys=False)['ep'].transform(lambda x: _trendPe(x, n, shift))

    return _raw


# 计算单季度ROE及ROE_ttm
def roe():
    _sql1 = """
        with is_quarter as (
            select iq.ticker, iq.f_ann_date, iq.end_date, iq.n_income_attr_p
            from (
                select *, row_number() over (
                    partition by ticker, end_date
                    order by update_flag desc, f_ann_date
                ) as rn
                from quant_research."financials_incomeState_quarter"
            ) iq
            where iq.rn=1
        ),
        bs_lt as (
            select bs.ticker, bs.f_ann_date, bs.end_date, bs.total_assets, bs.total_liab
            from (
                select *, row_number() over (
                    partition by ticker, end_date
                    order by update_flag desc, f_ann_date
                ) as rn
                from quant_research.financials_bs_lt
            ) bs
            where bs.rn=1
        )
        
        select a.ticker as code, a.f_ann_date, a.end_date, a.n_income_attr_p, (b.total_assets - b.total_liab) as net_assets
        from is_quarter as a
        left join bs_lt as b
            on a.ticker=b.ticker and a.end_date=b.end_date
        order by a.end_date
    """

    _raw = pd.read_sql(_sql1, engine)
    _raw_duplicates = _raw[_raw.duplicated(subset=['code', 'end_date'], keep=False)]
    if len(_raw_duplicates) != 0:
        raise ValueError("有重复数据，请检查")

    # 计算ROE
    def _roe(group):
        if len(group) < 2:
            return pd.Series(np.nan, index=group.index)
        else:
            _slice = sliding_window(group, 2, output='array')
            _r = [_s[-1][0] / np.mean([_s[0][1], _s[1][1]]) for _s in _slice]
            _rr = pd.Series(np.nan, index=group.index)
            _rr.iloc[1:] = _r
        return _rr

    def _roe_ttm(group):
        if len(group) < 4:
            return pd.Series(np.nan, index=group.index)
        else:
            _slice = sliding_window(group, 4, output='array')
            _r = [np.sum(_s[:, 0]) / np.mean(_s[:, 1]) for _s in _slice]
            _rr = pd.Series(np.nan, index=group.index)
            _rr.iloc[3:] = _r
        return _rr

    _raw['roe_q'] = _raw.groupby('code', group_keys=False)[['n_income_attr_p', 'net_assets']].apply(lambda x: _roe(x))
    # _raw['roe_ttm'] = _raw.groupby('code', group_keys=False)[['n_income_attr_p', 'net_assets']].apply(
    #     lambda x: _roe_ttm(x))
    _raw.dropna(inplace=True)
    # # 对齐交易日期
    # _signal = align_trade_date(_raw, engine, signal_columns=['roe_q', 'roe_ttm'])
    # _signal['trading'] = pd.to_datetime(_signal['trading']).dt.date
    # _signal['end_date'] = pd.to_datetime(_signal['end_date']).dt.date
    # # 写入数据库
    # bulk_insert_with_orm(_signal, 'factor_roe', engine)
    return _raw


# 研发费用：rd_exp, 合同负债：contract_liab，存货: inventories, 现金总资产比例 n_cashflow_act,
def financial_indicators():
    """
        计算研发费用率
    """
    _sql = """
    -- 利润表特殊指标
    with is_quarter as (
        select iq.ticker, iq.f_ann_date, iq.end_date, iq.rd_exp, iq.revenue
            from (
                select *, row_number() over (
                    partition by ticker, end_date
                    order by update_flag desc, f_ann_date
                ) as rn
                from quant_research."financials_incomeState_quarter"
            ) iq
            where iq.rn=1 and iq.revenue>0
        ),
    -- 资产负债表特殊指标
    bs_lt as (
        select bs.ticker, bs.f_ann_date, bs.end_date, bs.contract_liab, bs.inventories, bs.total_assets
            from (
                select *, row_number() over (
                    partition by ticker, end_date
                    order by update_flag desc, f_ann_date
                ) as rn
                from quant_research.financials_bs_lt
            ) bs
            where bs.rn=1
        ),
    -- 现金流量表特殊指标
    cf_lt as (
        select cf.ticker, cf.f_ann_date, cf.end_date, cf.n_cashflow_act
            from (
                select *, row_number() over (
                    partition by ticker, end_date
                    order by update_flag desc, f_ann_date
                ) as rn
                from quant_research.financials_cashflow_lt
            ) cf
            where cf.rn=1
    )
    -- 查询逻辑
    select a.ticker as code, a.f_ann_date, a.end_date, a.rd_exp/a.revenue as rd_ratio, 
        b.contract_liab/a.revenue as contract_liab_proportion, b.inventories/a.revenue as invent_proportion,
            c.n_cashflow_act/b.total_assets as cash_proportion
    from is_quarter a
    left join bs_lt b on a.ticker=b.ticker and a.end_date=b.end_date
    left join cf_lt c on a.ticker=c.ticker and a.end_date=c.end_date
    where a.end_date >= '2018-03-31'
    order by a.end_date
    """
    _raw = pd.read_sql(_sql, engine)

    # 将因子ttm化
    _raw['rd_ratio_ttm'] = _raw['rd_ratio'].rolling(4).mean()
    _raw['contract_liab_proportion_ttm'] = _raw['contract_liab_proportion'].rolling(4).mean()
    _raw['invent_proportion_ttm'] = _raw['invent_proportion'].rolling(4).mean()
    _raw['cash_proportion_ttm'] = _raw['cash_proportion'].rolling(4).mean()
    return _raw


# 季度利润同比、环比增速因子
def profit_growth():
    _sql = """
        -- 扣非归母净利润
        with is_quarter as (
            select iq.ticker, iq.f_ann_date, iq.end_date, iq.n_income_attr_p
            from (
                select *, row_number() over (
                    partition by ticker, end_date
                    order by update_flag desc, f_ann_date
                ) as rn
                from quant_research."financials_incomeState_quarter"
            ) iq
            where iq.rn=1
        ),
        -- 计算同比和环比增长率
        growth_rates as (
            select 
                ticker,
                f_ann_date,
                end_date,
                n_income_attr_p,
                -- 计算环比增长率（与上一期比较）
                case 
                    when lag(n_income_attr_p, 1) over (partition by ticker order by end_date) <= 0 
                        or n_income_attr_p <= 0 then null
                    else (n_income_attr_p - lag(n_income_attr_p, 1) over (partition by ticker order by end_date)) 
                        / lag(n_income_attr_p, 1) over (partition by ticker order by end_date)
                end as qoq_growth,
                -- 计算同比增长率（与去年同期比较）
                case 
                    when lag(n_income_attr_p, 4) over (partition by ticker order by end_date) <= 0 
                        or n_income_attr_p <= 0 then null
                    else (n_income_attr_p - lag(n_income_attr_p, 4) over (partition by ticker order by end_date)) 
                        / lag(n_income_attr_p, 4) over (partition by ticker order by end_date)
                end as yoy_growth
            from is_quarter
        )
        select 
            ticker as code,
            f_ann_date,
            end_date,
            n_income_attr_p,
            qoq_growth,
            yoy_growth
        from growth_rates
        where end_date>='2015-03-31'
        order by ticker, end_date
    """
    _raw = pd.read_sql(_sql, engine)
    return _raw


# 所有财务因子
def financial_factors():
    _sql = """
        -- 扣非归母净利润
        with is_quarter as (
            select iq.ticker, iq.f_ann_date, iq.end_date, iq.n_income_attr_p, 
                iq.rd_exp/nullif(iq.revenue,0) as rd_ratio, iq.revenue
            from (
                select *, row_number() over (
                    partition by ticker, end_date
                    order by update_flag desc, f_ann_date
                ) as rn
                from quant_research."financials_incomeState_quarter"
            ) iq
            where iq.rn=1
        ),
        bs_lt as (
            select bs.ticker, bs.f_ann_date, bs.end_date, bs.total_assets, bs.total_liab, bs.contract_liab
            from (
                select *, row_number() over (
                    partition by ticker, end_date
                    order by update_flag desc, f_ann_date
                ) as rn
                from quant_research.financials_bs_lt
            ) bs
            where bs.rn=1
        ),
        -- 现金流量表特殊指标
        cf_lt as (
            select cf.ticker, cf.f_ann_date, cf.end_date, cf.n_cashflow_act
                from (
                    select *, row_number() over (
                        partition by ticker, end_date
                        order by update_flag desc, f_ann_date
                    ) as rn
                    from quant_research.financials_cashflow_lt
                ) cf
                where cf.rn=1
        ),
        -- 计算同比和环比增长率
        growth_rates as (
            select 
                ticker,
                f_ann_date,
                end_date,
                n_income_attr_p,
                -- 计算同比增长率（与去年同期比较）
                case 
                    when lag(n_income_attr_p, 4) over (partition by ticker order by end_date) <= 0 
                        or n_income_attr_p <= 0 then null
                    else (n_income_attr_p - lag(n_income_attr_p, 4) over (partition by ticker order by end_date)) 
                        / lag(n_income_attr_p, 4) over (partition by ticker order by end_date)
                end as yoy_growth
            from is_quarter
        ),
        -- 计算净资产并添加上期净资产
        net_assets_with_lag as (
            select 
                a.ticker as code, 
                a.f_ann_date, 
                a.end_date, 
                a.n_income_attr_p, 
                (b.total_assets - b.total_liab) as net_assets,
                lag(b.total_assets - b.total_liab) over (
                    partition by a.ticker 
                    order by a.end_date
                ) as prev_net_assets
            from is_quarter as a
            left join bs_lt as b
                on a.ticker=b.ticker and a.end_date=b.end_date
        ),
        -- 计算ROE_Q
        roe_quarter as (
            select 
                code, 
                f_ann_date, 
                end_date, 
                n_income_attr_p, 
                net_assets,
                prev_net_assets,
                case 
                    when prev_net_assets is not null and (prev_net_assets + net_assets) != 0 
                    then n_income_attr_p / ((prev_net_assets + net_assets) / 2)
                    end as roe_q
            from net_assets_with_lag
        )
        
        -- 输出结果
        select 
            a.ticker as code,
            a.f_ann_date,
            a.end_date,
            a.yoy_growth as net_income_yoy,
            b.roe_q,
            c.rd_ratio,
            d.contract_liab/nullif(c.revenue, 0) as contract_proportion,
            e.n_cashflow_act/nullif(d.total_assets,0) as cashflow_proportion
        from growth_rates as a
        left join roe_quarter as b
            on a.ticker=b.code and a.end_date=b.end_date
        left join is_quarter as c
            on a.ticker=c.ticker and a.end_date=c.end_date
        left join bs_lt as d
            on a.ticker=d.ticker and a.end_date=d.end_date
        left join cf_lt as e
            on a.ticker=e.ticker and a.end_date=e.end_date
        where a.end_date>='2015-03-31'
        order by a.ticker, a.end_date
    """

    raw = pd.read_sql(_sql, engine)
    bulk_insert_with_orm(raw, 'factor_financial', engine)
    return raw


"""
    论文因子研究：
    1. PG因子：营运利润同比增长量/去年同期账面权益价值
        - 测试效果：分层效果好(因子值-行业均值，行业中性化效果差不多)
    2. 基本面动量因子：SUE, CAR因子
        - SUE因子：效果持续好
        - CAR 因子无选股效果
        - 策略：SUE，与长期动量共振
    3. 盈利质量+成长结合因子
    4. 研发/无形资产相关信号
    ...
    探索财务因子有效时间
    
    研报复现：综合成长因子
"""


def factor_pg(is_industry: bool = False):
    """
    PG因子=营运利润同比增长/去年同期账面权益价值
    - 营运利润=营业收入-营业成本-销售费用-管理费用-利息费用
    """
    _sql = """
        with is_quarter as (
            select iq.ticker, iq.f_ann_date, iq.end_date, iq.operate_profit
            from (
                select *, row_number() over (
                    partition by ticker, end_date
                    order by update_flag desc, f_ann_date
                ) as rn
                from quant_research."financials_incomeState_quarter"
            ) iq
            where iq.rn=1),
            bs_quarter as (
                select bs.ticker, bs.f_ann_date, bs.end_date, bs.total_hldr_eqy_exc_min_int as holder_equity
                from (
                    select *, row_number() over (
                        partition by ticker, end_date
                        order by update_flag desc, f_ann_date
                    ) as rn
                    from quant_research."financials_bs_lt"
                ) bs
                where bs.rn=1),
            combined_data as (
                select 
                    iq.ticker as code, 
                    iq.f_ann_date, 
                    iq.end_date, 
                    iq.operate_profit,
                    bs.holder_equity,
                    lag(iq.operate_profit, 4) over (partition by iq.ticker order by iq.end_date) as prev_operate_profit,
                    lag(bs.holder_equity, 4) over (partition by iq.ticker order by iq.end_date) as prev_holder_equity
                from is_quarter as iq
                left join bs_quarter as bs
                    on iq.ticker=bs.ticker and iq.end_date=bs.end_date
            )
        select 
            a.code, 
            a.f_ann_date, 
            a.end_date, 
            (a.operate_profit - a.prev_operate_profit) / a.prev_holder_equity as factor_pg,
            b.l1_name as industry_code
        from combined_data as a
        left join quant_research."sw_industry_constituent" as b
            on a.code=b.ts_code
        where a.prev_holder_equity is not null and a.prev_holder_equity != 0
        order by a.end_date
    """
    raw = pd.read_sql(_sql, engine)

    if is_industry:
        # 因子值减去行业均值
        raw['pg_industry'] = raw.groupby(['industry_code', 'end_date'])['factor_pg'].transform(lambda x: x - x.mean())

    return raw


def car_n(n: int = 5):
    """
    公司业绩披露前n日异常收益率
        - 业绩预告披露后
            * 涨跌幅：无选股效果
        - 业绩报告披露后5日
            * 涨跌幅：无选股效果

    """
    # 业绩公告前5日收益率
    _sql = """
        select iq.ticker as code, iq.ann_date as f_ann_date, iq.end_date, b.is_open, b.pretrade_date,
            case when b.is_open = 1 then iq.ann_date
            else b.pretrade_date
            end as car_end, c.l1_name as industry_code
        from (
                select *, row_number() over (
                    partition by ticker, end_date
                    order by update_flag desc, ann_date
                ) as rn
                from quant_research."financials_forecast"
            ) iq
        left join quant_research.basic_trading_date as b on iq.ann_date=b.cal_date
        left join quant_research.sw_industry_constituent as c on iq.ticker = c.ts_code
        where iq.rn=1
        order by iq.end_date;
    """
    _sql2 = """
        select * from quant_research.basic_trading_date where is_open=1
    """
    tradings = pd.read_sql(_sql2, engine)

    raw = pd.read_sql(_sql, engine)
    n = 0 - n - 1
    raw['car_start'] = raw['car_end'].apply(lambda x: get_next_tradeDate(x, tradings, n=n))

    # 获取前复权行情
    _sql3 = """
    select code, trade_date, close from quant_research.market_daily_qfq where trade_date > '2014-01-01'
    order by trade_date;
    """
    price = pd.read_sql(_sql3, engine)

    # 左连接行情
    raw = raw.merge(price, left_on=['code', 'car_start'], right_on=['code', 'trade_date'], how='left')
    raw.drop(columns=['trade_date'], inplace=True)
    raw.rename(columns={'close': 'close_start'}, inplace=True)

    raw = raw.merge(price, left_on=['code', 'car_end'], right_on=['code', 'trade_date'], how='left')
    raw.drop(columns=['trade_date'], inplace=True)
    raw.rename(columns={'close': 'close_end'}, inplace=True)

    # 计算收益率
    raw.dropna(inplace=True)
    raw['signal'] = raw['close_end'] / raw['close_start'] - 1
    return raw


# 业绩超预期因子
def sue():
    """
        基于实际披露的财务数据计算财务超预期因子，预期值采用
        预测值生成方法：
            * 简单随机游走模型：

    """
    # 获取历史财务数据
    _sql2 = """
    select ticker as code, end_date, f_ann_date, revenue, n_income_attr_p
    from (
        select *, row_number() over (
            partition by ticker, end_date
            order by update_flag desc, f_ann_date
        ) as rn
        from quant_research."financials_incomeState_quarter"
    ) iq
    where iq.rn=1
    order by end_date;
    """

    row = pd.read_sql(_sql2, engine)
    _duplicates = row[row.duplicated(['code', 'end_date'], keep=False)]
    if len(_duplicates) > 0:
        raise ValueError('请检查，存在重复值')

    def _core(single):
        try:
            _draft = np.mean(np.diff(single[0:8]))
            _forecast = single.values[7] + _draft

            return single.values[8] - _forecast
        except Exception as e:
            print(e)

    def _calculateSuprise(df: pd.DataFrame, category='revenue'):

        _r = np.full(len(df), np.nan)
        if len(df) < 9:
            return pd.Series(_r, index=df.index)
        else:
            df.sort_values(by='end_date', ascending=True, inplace=True)  # 按报告期正序排列吧
            # 每9个数字构建一个滑动窗口
            signals = df.rolling(window=9)[category].apply(lambda x: _core(x))

        return pd.Series(signals, index=df.index)

    row['sue_revenue'] = row.groupby('code', group_keys=False)[['f_ann_date', 'end_date', 'revenue']].apply(
        lambda group: _calculateSuprise(
            group.copy()
        ))

    row['sue_profit'] = row.groupby('code', group_keys=False)[['f_ann_date', 'end_date', 'n_income_attr_p']].apply(
        lambda group: _calculateSuprise(
            group.copy(), category='n_income_attr_p'
        ))

    signal = row[['code', 'f_ann_date', 'end_date', 'sue_profit', 'sue_revenue']].copy()  # 处理因子信号
    signal.dropna(inplace=True)

    _duplicates1 = signal[signal.duplicated(['code', 'end_date'], keep=False)]


    # 对齐交易日日期
    # signals = align_trade_date(signal, engine)
    return signal


if __name__ == '__main__':
    # # CAT-20财务因子检验
    # raw = car_n(20)
    # raw.dropna(inplace=True)
    #
    # # rd_ratio行业中性因子回测
    # f_rd_ttm = FinancialFactor(factor=raw)  # 默认：20交易日调仓周期
    #
    # f_rd_ttm.group_report(signal_name='signal_neutral')
    # f_rd_ttm.plotting_group_return(signal_name='signal_neutral')
    start = datetime.datetime.now()
    a = bp_compound()
    end = datetime.datetime.now()
    print(f"任务耗时：{end - start}")
    print(a)
